package com.study.flink.window

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * Time Window Demo
  * 包括滚动窗口和滑动窗口
  * @author stephen
  * @date 2019-07-22 15:59
  */
object ScalaTimeWindowDemo {

  def main(args: Array[String]): Unit = {

    // 1 获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2 获取输入数据
    import org.apache.flink.api.scala._
    val dataStream = env.socketTextStream("localhost", 9999)

    // 3 Transformation
    val resultStream = dataStream
      .flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(_._1)
      // 滚动窗口：默认，每1分钟计算一次
      // .timeWindow(Time.minutes(1))
      // 滑动窗口：每1分钟计算最近5分钟的数据
      .timeWindow(Time.minutes(5), Time.minutes(1))
      .reduce((x, y) => (x._1, x._2 + y._2))

    // 4 输出
    resultStream.print()

    // 5 启动任务
    env.execute("Time Window Demo")

  }
}
